package cn.jn.lhm.part07;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @author lhm 2753914576@qq.com
 * @time 2024/10/17 - 下午9:31
 **/
public class PollerIO extends ServerThread {

    private final Server server;

    private final Queue<SocketChannel> socketChannelQueue;

    public PollerIO(String name, Server server) {
        super(name);
        this.server = server;
        this.socketChannelQueue = new LinkedBlockingQueue<>();
    }

    public void bind(SocketChannel socketChannel) {
        this.socketChannelQueue.offer(socketChannel);

        // 防止selector 阻塞在select方法上
        this.wakeupSelector();
    }

    @Override
    public void run() {
        while (!server.isStopped()) {
            doSelect();
            doAcceptConnections();
        }
        closeSelector();
    }

    private void doAcceptConnections() {
        SocketChannel socketChannel;
        while (!server.isStopped() && (socketChannel = socketChannelQueue.poll()) != null) {
            try {
                // 关注读事件，接受客户端的数据
                socketChannel.register(this.selector, SelectionKey.OP_READ);
            } catch (ClosedChannelException e) {
                throw new RuntimeException(e);
            }
        }
    }

    private void doSelect() {
        try {
            // wakeupSelector() 就是防止这里被阻塞无法将新的连接绑定到 selector 上
            this.selector.select();
            Iterator<SelectionKey> iterator = this.selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                iterator.remove();
                if (!key.isValid()) {
                    continue;
                }
                if (key.isReadable()) {
                    // 读事件，处理客户端的数据
                    doRead(key);
                }
            }

        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private void doRead(SelectionKey key) {
        if (key.isReadable()) {
            SocketChannel socketChannel = (SocketChannel) key.channel();
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            try {
                int num = socketChannel.read(buffer);
                if (num > 0) {
                    // work pool 处理业务
                    System.out.println("read data from client: " + new String(buffer.array(), 0, num));
                    BuTask task = new BuTask(buffer, socketChannel);
                    this.server.addBuTask(task);
                }
            } catch (IOException e) {
                try {
                    socketChannel.close();
                } catch (IOException ex) {
                    throw new RuntimeException(ex);
                }
            }
        }
    }
}
